package com.xueyuan.wata.daph.flink117.api.node

import com.xueyuan.wata.daph.api.node.Node
import com.xueyuan.wata.daph.flink117.computer.FlinkComputer
import com.xueyuan.wata.daph.flink117.utils.CatalogUtil.registerCatalog
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.catalog.Catalog

trait FlinkNode extends Node {
  def computer: FlinkComputer

  def tableEnv: TableEnvironment = computer.tableEnv

  final def isEnableGlobalCatalog(catalogName: String): Boolean = computer.globalCatalogConfig(catalogName)("enabled").toBoolean

  final def registerGlobalCatalog(catalogName: String): Unit = {
    registerCatalog(tableEnv, catalogName, globalCatalog(catalogName))
  }

  final def useGlobalCatalog(catalogName: String): Unit = {
    tableEnv.useCatalog(catalogName)
  }

  final def globalCatalog(catalogName: String): Catalog = computer.globalCatalog(catalogName)
}
